package com.strato.queue.v1_0_0.exec.annotation.handler;

import java.lang.annotation.Annotation;

import org.osgi.framework.Bundle;

import com.strato.annotation.base.v1_0_0.handler.AbstractAnnotationHandler;
import com.strato.annotation.base.v1_0_0.registry.ClassDescriptor;
import com.strato.annotation.osgi.v1_0_0.service.util.ServiceCreator;
import com.strato.base.env.v1_0_0.GlobalEnvironment;
import com.strato.batch.base.v1_0_0.api.JobConsumer;
import com.strato.batch.base.v1_0_0.api.JobConsumerFactory;
import com.strato.batch.base.v1_0_0.batch.BatchExecutor;
import com.strato.batch.base.v1_0_0.batch.BatchJobExecutionConfig;
import com.strato.queue.v1_0_0.exec.ConsumerExecution;
import com.strato.queue.v1_0_0.exec.ExecutionManager;
import com.strato.queue.v1_0_0.exec.RoundRobinQueueJobProducer;
import com.strato.queue.v1_0_0.exec.annotation.QueueConsumer;
import com.strato.queue.v1_0_0.jdbc.model.QueueMessage;

public class QueueConsumerAnnotationHandler extends AbstractAnnotationHandler{

	@Override
	public void handleClass(Bundle bundle, Class<?> clazz, Annotation annotation) {
		if (!QueueConsumer.class.equals(annotation.annotationType())) {
			return;
		}
		QueueConsumer anno=clazz.getAnnotation(QueueConsumer.class);
		String concurrency = GlobalEnvironment.getInstance().resolveString(anno.concurrency());
		String messageType = GlobalEnvironment.getInstance().resolveString(anno.messageType());
		int holdMillis = GlobalEnvironment.getInstance().resolveInteger(anno.holdMillis());
		String consumerName=clazz.getName();
		com.strato.queue.v1_0_0.api.QueueConsumer consumer = (com.strato.queue.v1_0_0.api.QueueConsumer) ServiceCreator.newInstance(bundle, clazz);
		{
			BatchExecutor<QueueMessage> executor=new BatchExecutor<QueueMessage>(consumerName);
			BatchJobExecutionConfig<QueueMessage> config=new BatchJobExecutionConfig<QueueMessage>();
			config.setConcurrency(Integer.parseInt(concurrency));
			RoundRobinQueueJobProducer producer=new RoundRobinQueueJobProducer(messageType,holdMillis);
			config.setJobConsumerFactory(new QueueConsumerFactory(consumer));
			ConsumerExecution<QueueMessage> execution=new ConsumerExecution<QueueMessage>();
			execution.setConfig(config);
			execution.setExecutor(executor);
			execution.setProducer(producer);
			ExecutionManager.getInstance().setConsumerExecution(consumerName, execution);
			boolean autoStart=GlobalEnvironment.getInstance().resolveBoolean(anno.autoStart());
			if(autoStart){
				startExecution(execution);
			}
		}
	}

	private void startExecution(ConsumerExecution<QueueMessage> execution) {
		new Thread(){
			@Override
			public void run() {
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				execution.start();
			}
		}.start();
	}

	public class QueueConsumerFactory implements JobConsumerFactory<QueueMessage>{
		
		private com.strato.queue.v1_0_0.api.QueueConsumer consumer;

		public QueueConsumerFactory(com.strato.queue.v1_0_0.api.QueueConsumer consumer){
			this.consumer=consumer;
		}
		
		@Override
		public JobConsumer<QueueMessage> obtainConsumer() {
			return consumer;
		}
	}
	
	
	@Override
	public void handleUnresolveClass(Bundle bundle, ClassDescriptor classDescriptor) {
		Annotation anno = classDescriptor.getAnnotation(QueueConsumer.class);
		if(anno==null){
			return;
		}
		ConsumerExecution<?> execution = ExecutionManager.getInstance().getConsumerExecution(classDescriptor.getClassName());
		if(execution!=null){
			ExecutionManager.getInstance().removeConsumerExecution(classDescriptor.getClassName());
			execution.stop();
		}
	}
}
